import org.apache.spark.sql.SparkSession

object traffic_detect_alltheway {


  object trajectoryTrack {
    def trackAssemb(tuple:(Any, Iterable[(Any, Any)])) = {
      val carid = tuple._1
      val listAfterSort =tuple._2.iterator.toList.sortBy(key=>key._2.asInstanceOf[String])
      var traceTimeList: List[String]= List.empty
      var traceMonitorsList: List[String] =  List.empty
      for (each <-listAfterSort){trajectoryTrack
        traceMonitorsList =  traceMonitorsList:+each._1.asInstanceOf[String]
        traceTimeList = traceTimeList:+each._2.asInstanceOf[String]
      }
      (carid,(traceTimeList,traceMonitorsList))
    }
  }

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder().appName("traffic_detection").getOrCreate()

    spark.sql("use traffic")
    val flow_rdd = spark.sql("select * from monitor_flow_action").rdd


    val  bccarid = flow_rdd.filter(row=>{row.get(1)=="0001"}).map((_.get(3))).collect()
    val traceroute = flow_rdd.filter(row=>{bccarid.contains(row.get(3))}).map(key=>{(key.get(3),(key.get(1),key.get(4)))}).groupByKey().map(trajectoryTrack.trackAssemb).saveAsTextFile("hdfs://nameservice1/tmp/traffic_value.txt")

  }

}
